
import aiomysql
import asyncio
import json
from datetime import datetime
from collections import OrderedDict
import logging
import aiomysql
from utils.logger import logger


class LawDB:

    def __init__(self):
        self.conn = None
        self.cursor = None

    async def init_conn(self):

        try:
            self.conn = await aiomysql.connect(host='192.168.2.244',
                                               port=3306,
                                               user='root',
                                               password='linkAge@12345',
                                               db='stock_exchange',
                                               charset='utf8mb4',
                                               loop=asyncio.get_event_loop(),
                                               connect_timeout=60  # 设置连接超时时间为
                                               )
        except aiomysql.MySQLError as e:
            logging.error(f"数据库连接失败: {e}")
            raise ConnectionError(f"无法连接到数据库: {e}")

        self.cursor = await self.conn.cursor()

    async def close(self):
        if self.cursor:
            try:
                await self.cursor.close()
            except Exception as e:
                logging.warning(f"关闭cursor时出错: {e}")
        if self.conn:
            try:
                await self.conn.close()
                await self.conn.wait_closed()  # 等待连接关闭完成
            except Exception as e:
                logging.warning(f"关闭连接时出错: {e}")

    # 查询表格转html数据库tabel
    async def select_table_html(self, path: str):
        select_sql = """
            SELECT html
            FROM table_html
            WHERE path = %s 
            """
        try:
            await self.cursor.execute(select_sql, (path))
            await self.conn.commit()  # 确保提交事务
            table_count = await self.cursor.fetchone()  #
            print("全局历史表 查询气泡 成功")
            return table_count[0]
        except aiomysql.IntegrityError as e:
            print(f"表格2html 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f"表格2html 唯一值 错误；{e}, 跳过..."
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f" 查询表格2html 失败: {e}")
            # 可以在这里添加额外的错误处理逻辑，比如日志记录等

    # 保存 标注表
    async def save_mark(self, file_name: str, data: dict):
        try:
            data_json = json.dumps(data)
            select_sql = """
                SELECT *
                FROM mark
                WHERE file_name = %s
                """
            await self.cursor.execute(select_sql, (file_name))
            await self.conn.commit()  # 确保提交事务
            mark_count = await self.cursor.fetchall()
            if len(mark_count) == 0:  # 插入
                insert_sql = """
                INSERT INTO mark (file_name, data) 
                VALUES (%s, %s)
                """
                await self.cursor.execute(insert_sql, (file_name, data_json))
                await self.conn.commit()  # 确保提交事务
            else:  # 更新
                update_sql = """
                UPDATE mark
                SET data = %s
                WHERE file_name = %s
                """
                await self.cursor.execute(update_sql, (data_json, file_name))
                await self.conn.commit()  # 确保提交事务
            print("标注表 存入 成功")
            return '标注表 存入 成功'
        except aiomysql.IntegrityError as e:
            print(f"标注表 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f"标注表 唯一值 错误；{e}, 跳过..."
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f"标注表 插入气泡 失败: {e}")
            # 可以在这里添加额外的错误处理逻辑，比如日志记录等

    # 取出 标注表
    async def get_mark(self, file_name: str):
        try:
            select_sql = """
                SELECT *
                FROM mark
                WHERE file_name = %s
                """
            await self.cursor.execute(select_sql, (file_name))
            await self.conn.commit()  # 确保提交事务
            mark_count = await self.cursor.fetchall()   # 取出所有数据
            if len(mark_count) == 0:
                return '未找到'
            else:
                mark_list = []
                for mark in mark_count:
                    mark_dict = {'mark_id': mark[0],
                                'file_name': mark[1],
                                'data': mark[2]}
                    mark_list.append(mark_dict)

                return mark_list
        except Exception as e:
            print(f"标注表 查询 错误；{e}")
            await self.conn.rollback()

    ##################################################################################################################################

    # 更新全局 距离
    async def update_overall_distance(self, user_id: int, file_name: str, bubble: dict, fileId: str, status_code: int,
                             slice_index: int, distance: float):
        # 获取当前时间
        now = datetime.now()
        # 格式化时间为字符串
        update_time_str = now.strftime("%Y-%m-%d %H:%M:%S")

        bubble_json = json.dumps(bubble)

        select_sql = """
            SELECT *
            FROM history_overall
            WHERE user_id = %s AND fileId = %s AND slice_index = %s
        """
        insert_sql = """
        INSERT INTO history_overall (user_id, file_name, bubble, fileId, status_code, slice_index, update_time, distance) 
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
        """
        update_sql = """
        UPDATE history_overall
        SET bubble = %s, update_time = %s, distance = %s
        WHERE user_id = %s AND fileId = %s AND slice_index = %s
        """
        try:
            await self.cursor.execute(select_sql, (
            user_id, fileId, slice_index))
            await self.conn.commit()  # 确保提交事务
            select_count = await self.cursor.fetchall()
            if len(select_count) == 0:
                await self.cursor.execute(insert_sql, (
                user_id, file_name, bubble_json, fileId, status_code, slice_index, update_time_str, distance))
                await self.conn.commit()  # 确保提交事务
                print("全局历史表 距离插入气泡 成功")
            else:
                await self.cursor.execute(update_sql, (
                bubble_json, update_time_str, distance, user_id, fileId, slice_index))
                await self.conn.commit()  # 确保提交事务
                print("全局历史表 距离更新气泡 成功")
            return '全局历史表 距离插入气泡 成功'
        except aiomysql.IntegrityError as e:
            print(f"全局历史表 距离唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f"全局历史表 距离唯一值 错误；{e}, 跳过..."
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f"全局历史表 距离插入气泡 失败: {e}")

    # 更新全局 分数
    async def update_overall_score(self, user_id: int, file_name: str, bubble: dict, fileId: str, status_code: int,
                                      slice_index: int, score: int):
        # 获取当前时间
        now = datetime.now()
        # 格式化时间为字符串
        update_time_str = now.strftime("%Y-%m-%d %H:%M:%S")

        bubble_json = json.dumps(bubble)

        select_sql = """
            SELECT *
            FROM history_overall
            WHERE user_id = %s AND fileId = %s AND slice_index = %s
        """
        insert_sql = """
        INSERT INTO history_overall (user_id, file_name, bubble, fileId, status_code, slice_index, update_time, score) 
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
        """
        update_sql = """
        UPDATE history_overall
        SET update_time = %s, score = %s
        WHERE user_id = %s AND fileId = %s AND slice_index = %s
        """
        try:
            await self.cursor.execute(select_sql, (
                user_id, fileId, slice_index))
            await self.conn.commit()  # 确保提交事务
            select_count = await self.cursor.fetchall()
            if len(select_count) == 0:
                await self.cursor.execute(insert_sql, (
                    user_id, file_name, bubble_json, fileId, status_code, slice_index, update_time_str, score))
                await self.conn.commit()  # 确保提交事务
                print("全局历史表 分数插入气泡 成功")
            else:
                await self.cursor.execute(update_sql, (
                    update_time_str, score, user_id, fileId, slice_index))
                await self.conn.commit()  # 确保提交事务
                print("全局历史表 分数更新气泡 成功")
            return '全局历史表 分数插入气泡 成功'
        except aiomysql.IntegrityError as e:
            print(f"全局历史表 分数唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f"全局历史表 分数唯一值 错误；{e}, 跳过..."
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f"全局历史表 分数插入气泡 失败: {e}")


    # 获取全局接表中 气泡中的 problem_id_list
    async def get_history_overall_bubble(self, user_id: id, fileId: str, slice_index: int):
        select_sql = """
                   SELECT bubble 
                   FROM history_overall
                   WHERE user_id = %s AND fileId = %s AND slice_index = %s
                   """
        try:
            await self.cursor.execute(select_sql, (user_id, fileId, slice_index))
            await self.conn.commit()  # 确保提交事务
            history_count = await self.cursor.fetchone()  #
            bubble_dict = json.loads(history_count[0])
            # problem_id_list = bubble_dict['problem_id_list']
            print("全局历史表 获取bubble 成功")
            return bubble_dict
        except aiomysql.IntegrityError as e:
            print(f"全局历史表 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f"全局历史表 唯一值 错误；{e}, 跳过..."
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f" 获取bubble 失败: {e}")


    # 预防最后一个批次没有出现气泡，导致进度条不为100%
    async def save_overall_progress_bar(self, user_id:int, fileId: str):

        select_sql = """
                    SELECT MAX(slice_index)
                    FROM history_overall
                    WHERE user_id = %s AND fileId = %s AND score >= 8
            """
        save_sql = """
                UPDATE history_overall
                SET bubble_num = 1
                WHERE user_id = %s AND fileId = %s AND score >= 8 AND slice_index = %s
            """
        try:
            await self.cursor.execute(select_sql, (user_id, fileId))
            await self.conn.commit()  # 确保提交事务
            select_comment = await self.cursor.fetchone()
            select_comment_id = select_comment[0]

            await self.cursor.execute(save_sql, (user_id, fileId, select_comment_id))
            await self.conn.commit()  # 确保提交事务
            return  '更新进度条成功'

        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f" 更新进度条 失败: {e}")


    # 查找任务完成的时间线
    async def select_task_timeline(self, fileId: str):
        sql = """ 
            SELECT created_time , CASE WHEN status = 'finished' THEN updated_time ELSE NULL END AS end_time FROM task where file_id = %s;
        """
        try:
            await self.cursor.execute(sql, (fileId))
            task_timeline = await self.cursor.fetchone()
            start_time, end_time = None, None
            if task_timeline:
                start_time = task_timeline[0]
                end_time = task_timeline[1]
                if start_time:
                    start_time = start_time.strftime("%Y-%m-%d %H:%M:%S")
                if end_time:
                    end_time = end_time.strftime("%Y-%m-%d %H:%M:%S")
            return {"fileId": fileId, "start_time": start_time, "end_time": end_time}
        except Exception as e:
            print(f"查询任务总耗时失败；{e}")
            await self.conn.rollback()
            return f"查询任务总耗时失败；{e}"

    async def select_rules_reference(self, fileId: str):
        sql = """select bubble from history_overall where fileId = %s and bubble like %s order by slice_index asc"""
        try:
            await self.cursor.execute(sql, (fileId, '%rule_ref%'))
            await self.conn.commit()
            history_list = await self.cursor.fetchall()
            rules_reference = []
            for history in history_list:
                bubble = history[0]
                bubble_dict = json.loads(bubble)
                for rule in bubble_dict['rule_ref']:
                    if rule not in rules_reference:
                        rules_reference.append(rule)
            return rules_reference
        except Exception as e:
            print(f"查询参考规则失败；{e}")
            await self.conn.rollback()
            return f"查询参考规则失败；{e}"



    # 查询 全局表 中需要展示的气泡
    # async def select_history_overall_show(self, user_id: int, fileId: str):
    async def select_history_overall_show(self, fileId: str):

        # 根据 score >= 8
        select_sql = """
            SELECT user_id, file_name, bubble, fileId, status_code, slice_index, update_time, score, bubble_num, distance
            FROM history_overall
            WHERE fileId = %s AND score >= 7
            ORDER BY update_time asc
            """
        try:
            await self.cursor.execute(select_sql, (fileId,))
            await self.conn.commit()  # 确保提交事务
            history_count = await self.cursor.fetchall()  #
            history_list = []
            for history in history_count:
                history_dict = {'user_id': history[0],
                                'file_name': history[1],
                                'bubble': history[2],
                                'fileId': history[3],
                                'status_code': history[4],
                                'slice_index': history[5],
                                'update_time': history[6],
                                'score': history[7],
                                'bubble_num': history[8],
                                'distance': history[9]}
                bubble = json.loads(history[2])
                if 'aiText' not in bubble or not bubble['aiText']:
                    continue
                history_list.append(history_dict)
            print("全局历史表 查询气泡 成功")
            return history_list
        except aiomysql.IntegrityError as e:
            print(f"全局历史表 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f"全局历史表 唯一值 错误；{e}, 跳过..."
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f" 查询气泡 失败: {e}")
            # 可以在这里添加额外的错误处理逻辑，比如日志记录等

    # 插入 全局表
    async def insert_history_overall(self, user_id: int, file_name: str, bubble: dict, fileId: str, status_code: int,
                             slice_index: int, score: int, distance: float):
        # 获取当前时间
        now = datetime.now()

        # 格式化时间为字符串
        update_time_str = now.strftime("%Y-%m-%d %H:%M:%S")

        bubble_json = json.dumps(bubble)
        insert_sql = """
        INSERT INTO history_overall (user_id, file_name, bubble, fileId, status_code, slice_index, update_time, score, distance) 
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
        """
        try:
            await self.cursor.execute(insert_sql, (
            user_id, file_name, bubble_json, fileId, status_code, slice_index, update_time_str, score, distance))
            await self.conn.commit()  # 确保提交事务
            print("全局历史表 插入气泡 成功")
            return '全局历史表 插入气泡 成功'
        except aiomysql.IntegrityError as e:
            print(f"全局历史表 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f"全局历史表 唯一值 错误；{e}, 跳过..."
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f"全局历史表 插入气泡 失败: {e}")
            # 可以在这里添加额外的错误处理逻辑，比如日志记录等
    # 删除 全局表
    async def delete_history_overall(self, user_id: int, fileId: str, slice_index: int):
        delete_sql = """
        DELETE FROM history_overall
        WHERE user_id = %s AND fileId = %s AND slice_index = %s;
        """
        try:
            await self.cursor.execute(delete_sql, (user_id, fileId, slice_index))
            await self.conn.commit()  # 确保提交事务
            print("全局历史表 删除气泡 成功")
            return '全局历史表 删除气泡 成功'
        except aiomysql.IntegrityError as e:
            print(f"全局历史表 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f'全局历史表 唯一值 错误；{e}, 跳过...'
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f"全局历史表 删除气泡 失败: {e}")
            # 可以在这里添加额外的错误处理逻辑，比如日志记录等

    # 保存气泡 到 全局表（前端用）
    async def save_history_overall(self, user_id: int, fileId: str, bubble: dict, status_code: int, slice_index: int):
        # 获取当前时间
        now = datetime.now()

        # 格式化时间为字符串
        update_time_str = now.strftime("%Y-%m-%d %H:%M:%S")

        bubble_json = json.dumps(bubble)
        update_sql = """
        UPDATE history_overall
        SET 
            bubble = %s, 
            status_code = %s,
            update_time = %s
        WHERE user_id = %s AND fileId = %s AND slice_index = %s
        """
        try:
            await self.cursor.execute(update_sql,
                                      (bubble_json, status_code, update_time_str, user_id, fileId, slice_index))
            await self.conn.commit()  # 确保提交事务
            print("全局历史表 保存气泡 成功")
            return '全局历史表 保存气泡 成功'
        except aiomysql.IntegrityError as e:
            print(f"全局历史表 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f'全局历史表 唯一值 错误；{e}, 跳过...'
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f"全局历史表 保存气泡 失败: {e}")
            # 可以在这里添加额外的错误处理逻辑，比如日志记录等


    # 更新 全局表(自用）
    async def update_history_overall(self, user_id: int, fileId: str, bubble: dict, status_code: int, slice_index: int, bubble_num: float):
        # 获取当前时间
        now = datetime.now()

        # 格式化时间为字符串
        update_time_str = now.strftime("%Y-%m-%d %H:%M:%S")

        bubble_json = json.dumps(bubble)
        update_sql = """
        UPDATE history_overall
        SET 
            bubble = %s, 
            status_code = %s,
            update_time = %s,
            bubble_num = %s
        WHERE user_id = %s AND fileId = %s AND slice_index = %s
        """
        try:
            await self.cursor.execute(update_sql,
                                      (bubble_json, status_code, update_time_str, bubble_num, user_id, fileId, slice_index))
            await self.conn.commit()  # 确保提交事务
            print("全局历史表 更新气泡 成功")
            return '全局历史表 更新气泡 成功'
        except aiomysql.IntegrityError as e:
            print(f"全局历史表 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f'全局历史表 唯一值 错误；{e}, 跳过...'
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f"全局历史表 更新气泡 失败: {e}")
            # 可以在这里添加额外的错误处理逻辑，比如日志记录等

    # 查询 全局表 user_id 和 fileId 下的数据
    async def select_history_overall(self, user_id: int, fileId: str):
        select_sql = """
            SELECT user_id, file_name, bubble, fileId, status_code, slice_index, update_time, score 
            FROM history_overall
            WHERE user_id = %s AND fileId = %s
            """
        try:
            await self.cursor.execute(select_sql, (user_id, fileId))
            await self.conn.commit()  # 确保提交事务
            history_count = await self.cursor.fetchall()  #
            history_list = []
            for history in history_count:
                history_dict = {'user_id': history[0],
                                'file_name': history[1],
                                'bubble': history[2],
                                'fileId': history[3],
                                'status_code': history[4],
                                'slice_index': history[5],
                                'update_time': history[6],
                                'score': history[7]}

                history_list.append(history_dict)

            print("全局历史表 查询气泡 成功")
            return history_list
        except aiomysql.IntegrityError as e:
            print(f"全局历史表 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f"全局历史表 唯一值 错误；{e}, 跳过..."
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f" 查询气泡 失败: {e}")
            # 可以在这里添加额外的错误处理逻辑，比如日志记录等

    # async def select_history_overall(self, user_id: int, fileId: str, lowest_score: int, highest_score: int, limit_num: int, offset_num: int):
    #     select_sql_all = """
    #         SELECT user_id, file_name, bubble, fileId, status_code, slice_index, update_time, score
    #         FROM history_overall
    #         WHERE user_id = %s AND fileId = %s AND score BETWEEN %s AND %s;
    #         """
    #     select_sql = """
    #         SELECT user_id, file_name, bubble, fileId, status_code, slice_index, update_time, score
    #         FROM history_overall
    #         WHERE user_id = %s AND fileId = %s AND score BETWEEN %s AND %s
    #         LIMIT %s OFFSET %s;
    #         """
    #     try:
    #         await self.cursor.execute(select_sql_all, (user_id, fileId, lowest_score, highest_score))
    #         await self.conn.commit()
    #         history_all = await self.cursor.fetchall()
    #         history_num = len(history_all)
    #
    #
    #         await self.cursor.execute(select_sql, (user_id, fileId, lowest_score, highest_score, limit_num, offset_num))
    #         await self.conn.commit()  # 确保提交事务
    #         history_count = await self.cursor.fetchall()
    #         history_list = [{"本次查询总条数": history_num}]
    #         for history in history_count:
    #             history_dict = {'user_id': history[0],
    #                             'file_name': history[1],
    #                             'bubble': history[2],
    #                             'fileId': history[3],
    #                             'status_code': history[4],
    #                             'slice_index': history[5],
    #                             'update_time': history[6],
    #                             'score': history[7]}
    #
    #             history_list.append(history_dict)
    #
    #         print("全局历史表 查询气泡 成功")
    #         return history_list
    #     except aiomysql.IntegrityError as e:
    #         print(f"全局历史表 唯一值 错误；{e}, 跳过...")
    #         await self.conn.rollback()
    #         return f"全局历史表 唯一值 错误；{e}, 跳过..."
    #     except Exception as e:
    #         await self.conn.rollback()  # 如果有错误发生，回滚事务
    #         raise ValueError(f" 查询气泡 失败: {e}")
    #         # 可以在这里添加额外的错误处理逻辑，比如日志记录等

    ################################################################################################################################

    # 插入 历史 气泡

    async def insert_history(self, user_id: int, file_name: str, bubble: dict, fileId: str, status_code: int, slice_index: int, batch_slice_index: str):
        # 获取当前时间
        now = datetime.now()

        # 格式化时间为字符串
        update_time_str = now.strftime("%Y-%m-%d %H:%M:%S")

        bubble_json = json.dumps(bubble)
        insert_sql = """
        INSERT INTO history (user_id, file_name, bubble, fileId, status_code, slice_index, update_time, batch_slice_index) 
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
        """
        try:
            await self.cursor.execute(insert_sql, (user_id, file_name, bubble_json, fileId, status_code, slice_index, update_time_str, batch_slice_index))
            await self.conn.commit()  # 确保提交事务
            print("历史表 插入气泡 成功")
            return '历史表 插入气泡 成功'
        except aiomysql.IntegrityError as e:
            print(f"历史表 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f"历史表 唯一值 错误；{e}, 跳过..."
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f"历史表 插入气泡 失败: {e}")
            # 可以在这里添加额外的错误处理逻辑，比如日志记录等

    # 删除 历史气泡
    async def delete_history(self, user_id: int, fileId: str, slice_index: int, batch_slice_index: str):

        try:
            delete_sql = """
            DELETE FROM history
            WHERE user_id = %s AND fileId = %s AND slice_index = %s AND batch_slice_index = %s;
            """
            await self.cursor.execute(delete_sql, (user_id, fileId, slice_index, batch_slice_index))

            await self.conn.commit()  # 确保提交事务
            print("历史表 删除气泡 成功")
            return '历史表 删除气泡 成功'
        except aiomysql.IntegrityError as e:
            print(f"历史表 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f'历史表 唯一值 错误；{e}, 跳过...'
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f"历史表 删除气泡 失败: {e}")
            # 可以在这里添加额外的错误处理逻辑，比如日志记录等

    # 更新 历史气泡
    async def update_history(self, user_id: int, fileId: str, bubble: dict, status_code: int, slice_index: int, batch_slice_index: str):
        # 获取当前时间
        now = datetime.now()
        # 格式化时间为字符串
        update_time_str = now.strftime("%Y-%m-%d %H:%M:%S")
        bubble_json = json.dumps(bubble)

        try:

            update_sql = """
            UPDATE history 
            SET 
                bubble = %s, 
                status_code = %s,
                update_time = %s
            WHERE user_id = %s AND fileId = %s AND slice_index = %s AND batch_slice_index = %s
            """
            await self.cursor.execute(update_sql, (bubble_json, status_code, update_time_str, user_id, fileId, slice_index, batch_slice_index))
            await self.conn.commit()  # 确保提交事务
            print("历史表 更新气泡 成功")
            return '历史表 更新气泡 成功'
        except aiomysql.IntegrityError as e:
            print(f"历史表 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f'历史表 唯一值 错误；{e}, 跳过...'
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f"历史表 更新气泡 失败: {e}")
            # 可以在这里添加额外的错误处理逻辑，比如日志记录等

    # 查询 历史表 user_id 和 fileId 下的数据
    async def select_history(self, user_id: int, fileId: str):
        select_sql = """
            SELECT user_id, file_name, bubble, fileId, status_code, slice_index, update_time, batch_slice_index 
            FROM history 
            WHERE user_id = %s AND fileId = %s
            """
        try:
            await self.cursor.execute(select_sql, (user_id, fileId))
            await self.conn.commit()  # 确保提交事务
            history_count = await self.cursor.fetchall()
            history_list = []
            for history in history_count:
                history_dict = {'user_id': history[0],
                                'file_name': history[1],
                                'bubble': history[2],
                                'fileId': history[3],
                                'status_code': history[4],
                                'slice_index': history[5],
                                'update_time': history[6],
                                'batch_slice_index': history[7]}
                history_list.append(history_dict)

            print("历史表 查询气泡 成功")
            return history_list
        except aiomysql.IntegrityError as e:
            print(f"历史表 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f"历史表 唯一值 错误；{e}, 跳过..."
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f"历史表 查询气泡 失败: {e}")
            # 可以在这里添加额外的错误处理逻辑，比如日志记录等

    # 更新 问题表的 问题分类 字段(智谱)
    async def update_problem_classification(self, problem_id: int, problem_classification: str):

        update_sql = """
        UPDATE problem_new 
        SET 
            problem_classification = %s
        WHERE problem_id = %s 
        """
        try:
            await self.cursor.execute(update_sql, (problem_classification, problem_id))
            await self.conn.commit()  # 确保提交事务
            print("问题表的 问题分类 字段 更新成功")
            return '问题表的 问题分类 字段 更新成功'
        except aiomysql.IntegrityError as e:
            print(f"问题表的 问题分类 字段 更新 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f'问题表的 问题分类 字段 更新 唯一值 错误；{e}, 跳过...'
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f"问题表的 问题分类 字段 更新 失败: {e}")
            # 可以在这里添加额外的错误处理逻辑，比如日志记录等

    # 更新 问题表的 问题分类 字段(V3)
    async def update_problem_classification_v3(self, problem_id: int, problem_classification: str):

        update_sql = """
        UPDATE problem_new 
        SET 
            problem_classification_v3 = %s
        WHERE problem_id = %s 
        """
        try:
            await self.cursor.execute(update_sql, (problem_classification, problem_id))
            await self.conn.commit()  # 确保提交事务
            print("问题表的 问题分类 字段 更新成功")
            return '问题表的 问题分类 字段 更新成功'
        except aiomysql.IntegrityError as e:
            print(f"问题表的 问题分类 字段 更新 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f'问题表的 问题分类 字段 更新 唯一值 错误；{e}, 跳过...'
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f"问题表的 问题分类 字段 更新 失败: {e}")
            # 可以在这里添加额外的错误处理逻辑，比如日志记录等
    #########################################################################
    # 查询 文本纠错表 中需要展示的气泡
    # async def select_correction_show(self, user_id: int, fileId: str):
    async def select_correction_show(self, fileId: str):

        select_sql = """
            SELECT user_id, bubble, fileId, slice_index, update_time, progress_bar
            FROM correction
            WHERE fileId = %s
            ORDER BY slice_index ASC
            """
        try:
            await self.cursor.execute(select_sql, (fileId))
            await self.conn.commit()  # 确保提交事务
            history_count = await self.cursor.fetchall()  #
            history_list = []
            for history in history_count:
                history_dict = {'user_id': history[0],
                                'bubble': history[1],
                                'fileId': history[2],
                                'slice_index': history[3],
                                'update_time': history[4],
                                'progress_bar': history[5]}
                history_list.append(history_dict)
            return history_list
            # if len(history_list) > 0:
            #     if int(history_list[-1]['progress_bar']) == 1:
            #         print("文本纠错表 查询气泡 成功， 进度100% 返回所有气泡")
            #         return history_list
            #     else:
            #         print("文本纠错表 查询气泡 成功， 进度未100% 返回空列表")
            #         return []
            # else:
            #     print("文本纠错表 查询气泡 成功， 未找到数据 返回空列表")
            #     return []
        except aiomysql.IntegrityError as e:
            print(f"文本纠错表 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f"文本纠错表 唯一值 错误；{e}, 跳过..."
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f" 文本纠错表 查询气泡 失败: {e}")
    # 可以在这里添加额外的错误处理逻辑，比如日志记录等


    # 分页查询project表中的上市审核项目
    async def select_project_show(self, page: int = 1, page_size: int = 10, company_name: str = "", company_abbreviation: str = "", status: str = "", stock_exchange: str = ""):
        offset = (page - 1) * page_size
        # 基础SQL查询 - 查询项目信息
        select_sql = """
        SELECT project_id, company_name, listing_process, company_abbreviation, audit_status, stock_exchange
        FROM project
        WHERE 1=1
        """
        params = []
        
        # 添加公司名称筛选条件
        if company_name:
            select_sql += " AND company_name LIKE %s"
            params.append(f"%{company_name}%")
        
        # 添加公司简称筛选条件
        if company_abbreviation:
            select_sql += " AND company_abbreviation LIKE %s"
            params.append(f"%{company_abbreviation}%")
        
        # 添加状态筛选条件
        if status:
            select_sql += " AND status = %s"
            params.append(status)
        
        # 添加交易所筛选条件
        if stock_exchange:
            select_sql += " AND stock_exchange = %s"
            params.append(stock_exchange)
        
        # 添加排序和分页
        select_sql += " ORDER BY project_id ASC LIMIT %s OFFSET %s"
        params.extend([page_size, offset])
        
        results = []
        try:
            await self.cursor.execute(select_sql, params)
            await self.conn.commit()  # 确保提交事务
            projects = await self.cursor.fetchall()
            
            # 如果没有查询到项目，直接返回空列表
            if not projects:
                return []
            
            # 提取所有project_id
            project_ids = [project[0] for project in projects]

            # 根据project_id查询对应的problem数据
            problem_sql = """
            SELECT project_id, problem_id, round, problem_order, problem_title
            FROM problem
            WHERE project_id IN (%s)
            ORDER BY project_id, round, problem_order ASC
            """
            
            # 构建IN子句的参数
            problem_params = list(project_ids)
            
            # 动态生成IN子句占位符
            placeholders = ', '.join(['%s'] * len(project_ids))
            problem_sql = problem_sql % placeholders

            logger.info(f"problem_sql: {problem_sql}")
            
            await self.cursor.execute(problem_sql, problem_params)
            problems = await self.cursor.fetchall()

            # 将problems按照project_id分组
            problems_by_project = {}
            # 用于披露保存problem_id
            problems_ids_by_project = {}
            for problem in problems:
                project_id = problem[0]
                if project_id not in problems_by_project:
                    problems_by_project[project_id] = []
                    problems_ids_by_project[project_id] = []

                problems_by_project[project_id].append({
                        'project_id': problem[0],
                        'problem_id': problem[1],
                        'round': problem[2],
                        'problem_order': problem[3],
                        'problem_title': problem[4]
                })
                problems_ids_by_project[project_id].append(problem[1])
            
            # 需要将每个project下面的problem聚合，然后根据problem_id_list去query表查找对应的content
            content_sql = """
            SELECT problem_id,  group_concat(query_content) as query_content_str
            FROM (
            SELECT problem_id, query_order, query_type, query_content
            FROM query q
            WHERE problem_id IN (%s)
            ORDER BY problem_id, query_order ASC
            ) A group by problem_id
            """

            query_content_map = {}

            for project_id, problem_ids in problems_ids_by_project.items():
                query_content_params = list(problem_ids)
                content_sql_str = content_sql % (', '.join(['%s'] * len(problem_ids)))
                await self.cursor.execute(content_sql_str, query_content_params)
                contents = await self.cursor.fetchall()
                # 这里需要存储下来具体的内容
                contents_dict = {}
                for content in contents:
                    contents_dict[content[0]] = content[1]
                query_content_map[project_id] = contents_dict

            for project, problems in problems_by_project.items():
                for problem in problems:
                    problem_id = problem['problem_id']
                    problem['query_content'] = query_content_map.get(project, {}).get(problem_id, '')



            for project in projects:
                project_item = {}
                project_item['project_id'] = project[0]
                project_item['company_name'] = project[1]
                project_item['listing_process'] = project[2]
                project_item['company_abbreviation'] = project[3]
                project_item['audit_status'] = project[4]
                project_item['stock_exchange'] = project[5]
                project_item['problems'] = problems_by_project.get(project[0], [])
                results.append(project_item)
            
            return results
        except aiomysql.IntegrityError as e:
            print(f"project表 唯一值 错误；{e}, 跳过...")
            await self.conn.rollback()
            return f"project表 唯一值 错误；{e}, 跳过..."
        except Exception as e:
            await self.conn.rollback()  # 如果有错误发生，回滚事务
            raise ValueError(f"project表 查询项目 失败: {e}")
    